package com.zhang.gmall.realtime.app.dws;

import com.zhang.gmall.realtime.beans.ProvinceStatsSQL;
import com.zhang.gmall.realtime.utils.ClickHouseUtil;
import com.zhang.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @title: 地区主题统计
 * @author: zhang
 * @date: 2022/3/15 17:37
 */
public class ProvinceStatsSqlApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.获取表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 3.检查点设置
        //TODO 4.从kafka读取数据创建动态表，生成watermark和提取事件时间字段
        String topic = "dwm_order_wide_2022";
        String groupID = "provinceStatsSqlApp";
        tableEnv.executeSql("create table order_wide (" +
                " order_id BIGINT," +
                " province_id BIGINT," +
                " province_name STRING," +
                " province_area_code STRING," +
                " province_iso_code STRING," +
                " province_3166_2_code STRING," +
                " split_total_amount DOUBLE," +
                " create_time STRING ," +
                " rt as TO_TIMESTAMP(create_time)," +
                " WATERMARK FOR rt AS rt - INTERVAL '5' SECOND" +
                ") WITH (" + MyKafkaUtil.getKafkaDDl(topic, groupID) +
                ")");

        //tableEnv.sqlQuery("select * from order_wide").execute().print();

        //TODO 5.从动态表中查询数据
        Table table = tableEnv.sqlQuery(
                " select" +
                        "    DATE_FORMAT(TUMBLE_START(rt, INTERVAL '10' second),'yyyy-MM-dd HH:mm:ss') stt," +
                        "    DATE_FORMAT(TUMBLE_end(rt, INTERVAL '10' second),'yyyy-MM-dd HH:mm:ss') edt, " +
                        "    province_id, " +
                        "    province_name, " +
                        "    province_area_code, " +
                        "    province_iso_code, " +
                        "    province_3166_2_code," +
                        "    count(distinct order_id) order_count, " +
                        "    sum(split_total_amount) order_amount, " +
                        "    UNIX_TIMESTAMP()*1000 ts " +
                        " from order_wide " +
                        "group by " +
                        "    province_id, " +
                        "    province_name, " +
                        "    province_area_code, " +
                        "    province_iso_code, " +
                        "    province_3166_2_code, " +
                        "    TUMBLE(rt, INTERVAL '10' second)"
        );


        //TODO 6.将动态表转化为流，写入clickhouse
        tableEnv
                .toAppendStream(table, ProvinceStatsSQL.class).print("ProvinceStatsSqlApp");

        tableEnv
                .toAppendStream(table, ProvinceStatsSQL.class)
                .addSink(ClickHouseUtil.getSinkFunction(
                        "insert into  province_stats_2022  values(?,?,?,?,?,?,?,?,?,?)"
                ));

        //TODO 1.执行任务
        env.execute();
    }
}
